package com.doit.day04

import scala.collection.immutable
import scala.io.Source

object _04_电影案例 {
  def main(args: Array[String]): Unit = {
    //读文件
    val movie: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\movies.txt").getLines().toList
    val rating: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\ratings.txt").getLines().toList
    val user: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\users.txt").getLines().toList
    //key 是电影id，value是电影名称和电影类型组合在一起的二元元祖
    val movies: Map[String, (String, String)] = movie.map(line => {
      //1_Toy Story (1995)_Animation|Children's|Comedy
      val arr: Array[String] = line.split("_")
      (arr(0), (arr(1), arr(2)))
    }).toMap

    val ratings: List[(String, String, String, String)] = rating.map(line => {
      //1_1193_5_978300760
      val arr: Array[String] = line.split("_")
      //四元元祖
      //用户id  电影id    评分     时间戳
      (arr(0), arr(1), arr(2), arr(3))
    })

    //key 是唯一的user_id  value是针对user_id其他的一些属性
    val users: Map[String, (String, String, String, String)] = user.map(line => {
      //1_F_1_10_48067
      val arr: Array[String] = line.split("_")
      //user_id  性别    年龄     职业      邮编
      (arr(0), (arr(1), arr(2), arr(3), arr(4)))
    }).toMap

    xuqiu5(users,ratings)


  }

  //1.求每部电影的平均分，按照降序输出  ==> 我需要得是电影名称 + 平均分
  def xuqiu1(movies: Map[String, (String, String)],ratings: List[(String, String, String, String)])={
    //1求平均分 ==》 ratings 每部电影的分数  分数的总和/分数的个数
    //3相同的放在一起 ratings 获取出来的只能够是电影id 而我最终要的是电影名称  join  左表movie 表几千   rating100万
    //2分数的总和 我就需要先把所有相同电影的分数都拿出来放在一起  sum  / 个数size
    //4左表做hashmap   groupBy
    //key 电影id   value是相同电影id的一个集合，这里面有所有电影id相同的评分
    val movieIdAndRatings: Map[String, List[(String, String, String, String)]] = ratings.groupBy(_._2)
    movieIdAndRatings.map(tp=>{
      //电影id
      val key: String = tp._1
      //value是相同电影id的一个集合，这里面有所有电影id相同的评分
      val value: List[(String, String, String, String)] = tp._2
      val scores: List[Double] = value.map(tup => {
        ////1_1193_5_978300760
        tup._3.toDouble
      })
      //取map中获取元素的时候，可能电影id有对应的电影名称，也可能没有
      val movieName: String = movies.getOrElse(key, (null, null))._1
      (movieName,scores.sum/scores.size)
    }).toList
      .sortBy(_._2)
      .reverse
      .take(5)
      .foreach(println)
    /*
    1.先求平均分  ==》 100万groupBy  几千条了   遍历一下，然后get
    2.如果是先将id和name转换，100万取遍历一下   然后针对100万去groupBy
     */

  }

  def xuqiu2(movies: Map[String, (String, String)],ratings: List[(String, String, String, String)])={
    val res: Map[String, Array[String]] = movies.map(tp => {
      val movie_id: String = tp._1
      val movie_type: String = tp._2._2
      //Animation|Children's|Comedy
      val arr: Array[String] = movie_type.split("\\|")
      //seq(Animation,Children's,Comedy)
      (movie_id,arr)
    })

    val res2: List[(String, String)] = ratings.map(tp => {
      val arr: Array[String] = res.getOrElse(tp._2, Array("未知"))
      for (elem <- arr) yield {
        (elem, tp._3)
      }
    }).flatten

    res2.groupBy(_._1)
      .map(tp=>{(tp._1,tp._2.map(tup=>tup._2.toDouble).sum/tp._2.size)})
      .toList
      .sortBy(_._2)
      .reverse
      .take(5)
      .foreach(println)

  }

  //3.哪些年份的电影评分(平均分)最高，取最高的前五个年份
  def xuqiu3(movies: Map[String, (String, String)],ratings: List[(String, String, String, String)])={
    //1.解析出来年份
    //2.用电影id相等，将电影id用年份替换
    //3.按照年份分组，求分数的平均值
    //4.排序取前5
    val idAndYear: Map[String, String] = movies.map(tp => {
      val movie_name: String = tp._2._1
      //Toy Story (1995)
      val year: String = movie_name.substring(movie_name.length - 5, movie_name.length - 1)
      (tp._1, year)
    })

    ratings.map(tp=>{
      val year: String = idAndYear.getOrElse(tp._2, "9999")
      (year,tp._3)
    }).groupBy(_._1)
      .map(tp=>{
        (tp._1,tp._2.map(_._2.toDouble).sum/tp._2.size)
      }).toList
      .sortBy(_._2)
      .reverse
      .take(5)
      .foreach(println)



  }

  //4.每个职业最喜欢的前三个电影id   观看次数最多的
  //每个人看到的电影是不是都在ratings里面 user_id 可以获取到job吗？
  //对职业进行分组 ==》 对user这份数据 读出来，做处理  ()
  //让他和ratings这张表进行join
  def xuqiu4(users: Map[String, (String, String, String, String)],ratings: List[(String, String, String, String)])={
    //为了将每一个user转换成职业
    //key 是职业   value 是职业和电影id的集合
    val map: Map[String, List[(String, String)]] = ratings.map(tp => {
      val user_id: String = tp._1
      val job_id: String = users.getOrElse(user_id, (null, null, null, null))._3
      (job_id, tp._2)
    }).groupBy(_._1)

    map.map(tp=>{
      val job_id: String = tp._1
      //List((10,1),(10,1),(10,2),(10,4))
      val jobAndMovieId: List[(String, String)] = tp._2
      val movieIdS: List[String] = jobAndMovieId.map(_._2)
      //WORDCOUNT
      val tuples: List[(String, Int)] = movieIdS.groupBy(word => word)
        .map(tp => (tp._1, tp._2.size))
        .toList
        .sortBy(_._2)
        .reverse
        .take(3)
      (job_id,tuples)
    }).foreach(println)
  }

  //5.年龄段在“18-24”的男性年轻人，最喜欢看哪10部电影（Id）
  //取统计次数的时候 ==》 100万条数据中，应该有很多数据都是需要过滤掉的  ratings   用户id 电影id
  //  collect  ==》  类似于什么？filter 和 map                    user表    M   18
  //先join 再去过滤
  def xuqiu5(users: Map[String, (String, String, String, String)],ratings: List[(String, String, String, String)])={
    val movieID_Gender_Age: List[(String, String, String)] = ratings.map(tp => {
      val tuple: (String, String, String, String) = users.getOrElse(tp._1, (null, null, null, null))
      val gender: String = tuple._1
      val age: String = tuple._2
      (tp._2, gender, age)
    })

    //过滤出来有些不符合条件的人   M   18
    val filtered: List[(String, String, String)] = movieID_Gender_Age.filter(tp => "M".equals(tp._2) && "18".equals(tp._3))

    filtered.groupBy(_._1)
      .map(tp=>(tp._1,tp._2.size))
      .toList
      .sortBy(_._2)
      .reverse
      .take(10)
      .foreach(println)


  }









}
